package com.winner.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * 监听服务器上的kafka是否有相关的消息发过来
 */
@Component
@Slf4j
public class UserLogConsumer {

    /**
     * 定义此消费者接收topics = "demo"的消息，与controller中的topic对应上即可
     *
     * @param consumerRecord 变量代表消息本身
     */
    @KafkaListener(topics = {"user-log"})
    public void consumer(ConsumerRecord<String, String> consumerRecord) {
        //判断是否为null
        String value = consumerRecord.value();
        String topic = consumerRecord.topic();
        long offset = consumerRecord.offset();

        System.err.println("消费消息:" + value);
        System.err.println("消费topic:" + topic);
        System.err.println("消费offset:" + offset);
    }
}
